package demo;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

@Slf4j
public class MovieClassifyReducer extends Reducer<Text, DistanceAndLabel, Text, NullWritable> {
    private int k = 0;

    @Override
    protected void setup(Reducer<Text, DistanceAndLabel, Text, NullWritable>.Context context)
            throws IOException, InterruptedException {
        //初始化K值
        k = context.getConfiguration().getInt("K", 3);
    }

    /**
     * reduce阶段，setup函数初始化参数K值，reduce函数对相同键的值根据距离进行升序排序，
     * 取出前K个值，输出reduce读取进来的键和这K个值中类别的众数。
     * <p>
     * 输入：《测试数据，距离及训练数据的类别》，
     * 例如：967,1,18,0,43615,0,1,1,34,17,2,0,0,3,18,13,2,4,1,3,58,4,1 -> 1.978135948331415E31,0
     * <p>
     * 输出： 性别分类  ，测试数据
     *
     * @param key     测试数据，例如：1008,0,35,3,77064,0,5,4,3,3,3,3,0,7,14,3,2,0,3,2,36,0,3
     * @param value   距离及训练数据的类别
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<DistanceAndLabel> value,
                          Reducer<Text, DistanceAndLabel, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        String sex = getMost(getTopK(sort(value)));

        //0,1008,0,35,3,77064,0,5,4,3,3,3,3,0,7,14,3,2,0,3,2,36,0,3
        log.info("reduce阶段：{}", sex + "|" + key);
        context.write(new Text(sex + "," + key), NullWritable.get());
    }

    /**
     * 得到列表中类别的众数
     *
     * @param topK
     * @return
     */
    private String getMost(List<String> topK) {
        HashMap<String, Integer> labelTimes = new HashMap<>();
        for (String str : topK) {
            String label = str.substring(str.lastIndexOf(",") + 1, str.length());
            if (labelTimes.containsKey(label)) {
                labelTimes.put(label, labelTimes.get(label) + 1);
            } else {
                labelTimes.put(label, 1);
            }
        }

        int maxInt = Integer.MIN_VALUE;
        String mostLabel = "";
        for (Map.Entry<String, Integer> kv : labelTimes.entrySet()) {
            if (kv.getValue() > maxInt) {
                maxInt = kv.getValue();
                mostLabel = kv.getKey();
            }
        }
        return mostLabel;
    }

    /**
     * 取出列表中的前K个值
     *
     * @param sort
     * @return
     */
    private List<String> getTopK(List<String> sort) {
        return sort.subList(0, k);
    }

    /**
     * 根据距离升序排序
     *
     * @param value
     * @return
     */
    private List<String> sort(Iterable<DistanceAndLabel> value) {
        ArrayList<String> result = new ArrayList<>();
        for (DistanceAndLabel val : value) {
            result.add(val.toString());
        }
        String[] tmp = new String[result.size()];
        result.toArray(tmp);
        Arrays.sort(tmp, new Comparator<String>() {

            @Override
            public int compare(String o1, String o2) {
                double o1D = Double.parseDouble(o1.substring(0, o1.indexOf(",")));
                double o2D = Double.parseDouble(o2.substring(0, o2.indexOf(",")));
                if (o1D > o2D) {
                    return 1;
                } else if (o1D < o2D) {
                    return -1;
                } else {
                    return 0;
                }
            }
        });
        return Arrays.asList(tmp);
    }
}
